package com.xncoding.pos.service.thread;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.core.util.RandomUtil;
import com.google.common.collect.Lists;
import com.xncoding.pos.async.AsyncException;
import com.xncoding.pos.dao.TbUserDao;
import com.xncoding.pos.model.TbUser;
import com.xncoding.pos.service.thread.original.MyThreadFromCallable;
import com.xncoding.pos.utils.DateUtils;
import com.xncoding.pos.utils.JsonUtils;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.AsyncResult;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.*;
import java.util.stream.Collectors;

/**
 * AsyncDemo
 *
 * @author XiongNeng
 * @version 1.0
 * @since 2018/1/25
 */
@Component
@Slf4j
public class AsyncTaskService {

    @Resource
    DateUtils dateUtils;

    @Resource
    TbUserDao tbUserDao;

    //可以shutdown，awaitTermination方便处理一些所有线程全部执行完才能够调用的服务接口等。
    @Resource
    ExecutorService executorService;

    @Resource
    Executor taskExecutor;

    //注入spring线程池 ,保留核心线程数量个线程继续等待连接，不能shutdown，awaitTermination
    @Resource
    ThreadPoolTaskExecutor threadPoolTaskExecutor;

    // 主线程和子线程的在事务上是相互隔离的，子线程的异常不会影响主线程的事务混滚与否
    // （倘若主线程不主动throw出异常，子线程即使抛出了异常也不会影响主线程的）
    // 结论:线程之间的事务完全隔离，异步线程内仍是可以调用异步

    @Async("defaultThreadPool")
    public void dealNoReturnTask() {

        log.info("返回值为void的异步调用开始:{},{}", Thread.currentThread().getName(), dateUtils.getStrNowTime());
        try {
            Thread.sleep(3000);
            log.info("线程里事务执行完毕");
        } catch (InterruptedException e) {
            log.error("线程执行失败:{}", e.getMessage());
        }

        log.info("返回值为void的异步调用结束" + dateUtils.getStrNowTime());
    }

    //调用的异步方法，不能为同一个类的方法（包括同一个类的内部类），简单来说，因为Spring在启动扫描时会为其创建一个代理类，
    // 而同类调用时，调用是在实体内执行的,所以和平常调用是一样的。

    // 其他的注解如@Cache等也是一样的道理，说白了，就是Spring的代理机制造成的。所以在开发中，最好把异步服务单独抽出一个类来管理


    // 在spring中像@Async和@Transactional、cache等注解本质使用的是动态代理，
    // 其实Spring容器在初始化的时候Spring容器会将含有AOP注解的类对象“替换”为代理对象（简单这么理解），
    // 那么注解失效的原因就很明显了，就是因为调用方法的是对象本身而不是代理对象，因为没有经过Spring容器，

    @Async("taskExecutor")
    public void doException() throws AsyncException {
        log.warn("有异常的异步调用开始" + Thread.currentThread().getName());
        throw new AsyncException("空指针异常了");
    }

    @Async
    public Future<String> futureTest(int i) {
        log.warn("Future异步调用开始" + Thread.currentThread().getName());

        Future<String> future;
        try {
            log.info("Future异步线程，开始执行：" + dateUtils.getStrNowTime());
            //模拟业务处理
            Thread.sleep(1000 * 3);

            future = new AsyncResult<String>("Future返回数据success:" + i);
        } catch (InterruptedException e) {
            future = new AsyncResult<String>("error");
        }
        log.info("Future异步线程，执行结束：" + dateUtils.getStrNowTime());
        return future;
    }

    /**
     * 带有返回值的异步方法,抛出异常
     * <p>
     * 异常调用返回Future
     * 对于返回值是Future，不会被AsyncUncaughtExceptionHandler处理，需要我们在方法中捕获异常并处理
     * 或者在调用方在调用Futrue.get时捕获异常进行处理
     *
     * @return
     */
    @Async
    public Future<String> futureExcept() {
        Future<String> future;
        log.warn("Future异步线程，异常情况：" + dateUtils.getStrNowTime());
        future = new AsyncResult<>(RandomUtil.randomNumbers(10));
        int i = 1 / 0;

        return future;
    }


    /**
     * 执行多个带返回值的任务，并取得多个返回值
     * 异步非阻塞获取并行任务执行结果
     * -----------哪个任务先执行完，哪个任务的返回值就先打印。
     * ----------在CompletionService接口中如果当前没有任务被执行完，则completionService.take().get()方法还是呈阻塞特性
     * 使用take()方法
     * 方法take()取得最先完成任务的future对象，谁执行的时间最短谁就最先返回。
     * <p>
     * 使用poll()方法
     * 方法poll的作用是获取并移除表示下一个已完成任务的future，如果不存在这样的任务，则返回null，方法poll无阻塞的效果。可以设置超时时间
     */
    public void completionServiceTest1() {
        /**
         * 线程池中任务的返回结果,放回执行结束的任务
         */
        CompletionService<TbUser> cService = new ExecutorCompletionService<>(executorService);

        int lenthSize = 5;

        for (int i = 0; i < lenthSize; i++) {
            // 向里面扔任务
            cService.submit(() -> {
                int millis = RandomUtil.randomInt(1000, 3000);
                Thread.sleep(2000);
                TbUser tbUser = new TbUser();
                tbUser.setId((long) millis);
                tbUser.setAge(20);
                tbUser.setUserName(RandomUtil.randomString(5));
                // TbUser tbUser = tbUserDao.queryById(all.get(finalI).getId());
                return tbUser;
            });

        }

        try {
            // 检查线程池任务执行结果
            //take()将导致等待，对于长耗时任务，将导致长时间等待，甚至无法预知的错误
            // poll 不等待，有结果就返回一个 Future 对象，否则返回 null,因为poll 是非阻塞执行的。
            // take 方获取并移除表示下一个已完成任务的 Future，如果目前不存在这样的任务，则等待，take 方法时阻塞执行的
            // 如果需要用到返回值建议用take

            for (int i = 0; i < lenthSize; i++) {
                // 检索并移除表示下一个已完成任务的 Future，如果目前不存在这样的任务，则等待。
                //这一行没有完成的任务就阻塞
                TbUser tbUser = cService.take().get();
                log.info("-take().get()----打印日志------{}", tbUser.toString());
            }

        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }

    }

    /**
     * 不需要返回值的场景，不要使用CompletionService的方法
     *
     * @return
     */
    public int completionServiceTest() {
        /**
         * 线程池中任务的返回结果,放回执行结束的任务
         */
        CompletionService<String> cService = new ExecutorCompletionService<>(executorService);

        int lenthSize = 5;

        for (int i = 0; i < lenthSize; i++) {
            // 向里面扔任务
            cService.submit(() -> {
                TbUser tbUser = new TbUser();
                tbUser.setId((RandomUtil.randomLong()));
                tbUser.setAge(20);
                tbUser.setUserName(RandomUtil.randomString(5));
                log.info("任务执行结束-----打印日志------{}", tbUser.toString());
                return null;
            });
        }

        // 所以，业务场景中不需要使用任务返回值的 ，不建议使用 CompletionService，假如使用了，记得一定要从阻塞队列中移除掉 task 执行结果，避免 OOM！
        return lenthSize;
    }


    // take() 获取任务阻塞，直到可以拿到任务为止。
    // poll() 获取任务不阻塞，如果没有获取到任务直接返回null。
    // poll(long timeout, TimeUnit unit) 带超时时间等待的获取任务方法（一般推荐使用这种）

    /**
     * 多任务执行-控制超时时间
     */
    @SneakyThrows
    public void completionForPollTest() {

        CompletionService<TbUser> cService = new ExecutorCompletionService<>(taskExecutor);
        // CompletionService<TbUser> cService = new ExecutorCompletionService<>(executorService);

        int lenthSize = 5;

        for (int i = 0; i < lenthSize; i++) {
            // 向里面扔任务
            cService.submit(() -> {
                log.info("CompletionService异步调用开始" + Thread.currentThread().getName());

                int millis = RandomUtil.randomInt(100, 1000);
                Thread.sleep(millis);
                TbUser tbUser = new TbUser();
                tbUser.setId((long) millis);
                tbUser.setAge(20);
                tbUser.setUserName(RandomUtil.randomString(5));
                return tbUser;
            });
        }

        //完成的任务
        List<TbUser> responseDataList = new ArrayList<>();
        //使用有参数的poll()方法，可以定义Timeout时长，从而可以实现更准确的时间窗口控制。我们可以用poll()方法替代take()方法。
        //获取线程返回值,2秒没有则返回空
        for (int i = 0; i < lenthSize; i++) {
            log.info("获取线程返回值-----打印日志-----");
            // poll将等待2秒，超过后将返回null
            Future<TbUser> take = cService.poll(500, TimeUnit.MILLISECONDS);
            if (Objects.nonNull(take)) {
                responseDataList.add(take.get());

                // if (take.isDone()) {
                //     try {
                //         TbUser user = take.get();
                //         log.info("获取线程执行结果:{}", user.getId());
                //     } catch (InterruptedException e) {
                //         e.printStackTrace();
                //     } catch (ExecutionException e) {
                //         e.printStackTrace();
                //     }
                // }else {
                //     take.cancel(true);
                // }

            }

        }

        responseDataList.forEach(tbUser -> {
            log.info("完成的任务----:{}", tbUser.toString());
        });
        // 判断是否存在超时接口
        if (lenthSize != responseDataList.size()) {
            log.info("存在超时接口----");
        }

    }


    /**
     * spring多线程---注入spring线程池,实现Callable接口（不常用）
     */
    public void threadTest3() {
        Future<String> submit = threadPoolTaskExecutor.submit(new MyThreadFromCallable(tbUserDao, 1L));

        log.info("异步调用开始submit-Future：" + Thread.currentThread().getName());


        try {
            //超时时间
            // Long s = submit.get(2L, TimeUnit.SECONDS);
            String s = submit.get();
            log.info("实现Callable接口-----结果------{}", s);

        } catch (InterruptedException | ExecutionException e) {
            log.warn("超时执行~~~~submit.get");
        }
        log.info("threadPoolTaskExecutor-----执行状态------{}", submit.isDone());

    }


    /**
     * spring多线程---Runnable
     */
    public void threadTest2() {

        executorService.execute(() -> {
            log.info("异步调用开始" + Thread.currentThread().getName());

            //业务逻辑
            try {
                TimeUnit.SECONDS.sleep(3);
                log.info("任务执行结束-----打印日志------{}", DateUtil.now());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
        log.info("Runnable-----结束----打印日志------");

    }

    /**
     * spring多线程----Callable
     */
    public void threadTest1() {

        Future<String> future = getFutureTest(1L);

        boolean done = future.isDone();
        log.info("执行状态-----打印日志------{}", done);
        try {
            String s = future.get();
            log.info("执行结果-----打印日志------{}", s);
        } catch (Exception e) {
            e.printStackTrace();
        }


        // 判定线程池任务是否结束
        boolean terminated = executorService.isTerminated();
        log.info("判定线程池任务是否结束-----打印日志------{}", terminated);

        // shutdown() 会立即销毁ExecutorService，它会让ExecutorServic停止接收新的任务，并等待现有任务全部执行完毕再销毁。
        // executorService.shutdown();


        // 在执行awaitTermination前需要执行shutdown方法,否则awaitTermination一定超时并且返回false
        // executorService.shutdown();
        // try {
        //     if (!executorService.awaitTermination(800, TimeUnit.MILLISECONDS)) {
        //         executorService.shutdownNow();
        //     }
        // } catch (InterruptedException e) {
        //     executorService.shutdownNow();
        // }
        // // 先停止接收任务，然后再等待一定的时间让所有的任务都执行完毕，如果超过了给定的时间，则立刻结束任务。
        // terminated = executorService.isTerminated();
        // log.info("判定线程池任务是否结束-----打印日志------{}", terminated);

    }

    private Future<String> getFutureTest(Long id) {
        return executorService.submit(() -> {
            log.info("异步调用开始" + Thread.currentThread().getName());
            Thread.sleep(3000);
            TbUser tbUser = tbUserDao.queryById(id);
            TbUser user = Optional.ofNullable(tbUser).orElse(new TbUser());
            log.info("getFutureTest.queryById()-----打印日志------{}", user.toString());
            return JsonUtils.toJson(user);
        });
    }

    private Callable<String> getFirstCallable() {
        Callable<String> callable = () -> {
            System.out.println("实现 First Callable接口来编写异步执行任务");
            Thread.sleep(1000);
            return "返回 First Callable 接口线程执行结果";
        };
        return callable;
    }

    /**
     * 多任务执行-原生方式
     */
    public void invokeTest1() {
        ArrayList<MyThreadFromCallable> tasks = getMyThreadFromCallables();

        //返回所有成功执行的结果：
        try {
            tasks.add(new MyThreadFromCallable(tbUserDao, 4L));
            List<Future<String>> futures = executorService.invokeAll(tasks, 3, TimeUnit.SECONDS);

            for (Future<String> future : futures) {
                Optional.ofNullable(future).ifPresent(stringFuture -> {
                    try {
                        log.info("invokeAll-----timeout------{}", stringFuture.get());
                    } catch (InterruptedException | ExecutionException e) {
                        e.printStackTrace();
                        log.warn("incident delay data,future get warn", e);
                    }
                });

            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    /**
     * 返回所有成功执行的结果
     */
    public void invokeTest3() {
        ArrayList<MyThreadFromCallable> tasks = getMyThreadFromCallables();

        try {
            List<Future<String>> futures = executorService.invokeAll(tasks);

            for (Future<String> future : futures) {

                if (Objects.nonNull(future)) {
                    try {
                        log.info("invokeTest3----invokeAll-----打印日志------{}", future.get());
                    } catch (ExecutionException e) {
                        e.printStackTrace();
                    }

                }

            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    /**
     * 并返回其中的一个成功返回的结果
     */
    public void invokeTest2() {
        ArrayList<MyThreadFromCallable> tasks = getMyThreadFromCallables();

        try {
            String s = executorService.invokeAny(tasks);
            log.info("invokeTest2-----invokeAny-----打印日志------{}", s);
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }

    }

    /**
     * 3秒执行完的方法 1+2
     *
     * @return
     */
    private ArrayList<MyThreadFromCallable> getMyThreadFromCallables() {
        MyThreadFromCallable threadFromCallable = new MyThreadFromCallable(tbUserDao, 1L);
        return Lists.newArrayList(threadFromCallable, new MyThreadFromCallable(tbUserDao, 2L));
    }


    /**
     * 单个任务的异步-推荐使用这种方式
     */
    public void threadPoolTaskExecutorTest() {

        threadPoolTaskExecutor.execute(() -> {
            log.info("ThreadPoolTaskExecutor异步调用开始：" + Thread.currentThread().getName());

            //业务逻辑
            try {
                TimeUnit.SECONDS.sleep(3);
                log.info("execute-----打印日志------{}", DateUtil.now());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });

        Future<Integer> future = threadPoolTaskExecutor.submit(() -> {
            //业务逻辑
            try {
                TimeUnit.SECONDS.sleep(3);
                log.info("submit-----打印日志------{}", DateUtil.now());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return 1;
        });

        boolean done = future.isDone();
        log.info("future.isDone-----打印日志------{}", done);
        try {
            Integer integer = future.get();
            log.info("future.get()-----打印日志------{}", integer);
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }

    }


    /**
     * 等待所有任务完成--没有返回值
     */
    public void allDone3() {
        // 定义任务数量
        int taskCount = 10;
        log.info("任务开始----打印日志------{}", DateUtil.now());

        // bo TODO 2025/8/21 17:59 注意参数传递问题,,,

        // 创建并执行所有任务
        List<CompletableFuture<Void>> futures = new ArrayList<>();
        for (int i = 0; i < taskCount; i++) {
            final int index = i;
            futures.add(CompletableFuture.runAsync(() -> {
                try {
                    Thread.sleep(1000);
                    if (index == 2) {
                        throw new RuntimeException("任务 " + index + " 异常");
                    }
                    System.out.println("任务 " + index + " 完成");
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }, threadPoolTaskExecutor));
        }

        // 等待所有任务完成
        CompletableFuture<Void> allFutures = CompletableFuture.allOf(
                futures.toArray(new CompletableFuture[0])
        );

        // 处理所有任务完成后的操作
        allFutures.thenRun(() -> {
            System.out.println("所有任务已完成");
        }).join();  // 阻塞主线程直到所有任务完成

        log.info("任务都完成-----打印日志------{}", DateUtil.now());
    }


    /**
     * thenCombine："等两个任务都完成，再合并结果"
     */
    private void thenCombineTest() {
        // 1. 异步查询商品价格（模拟耗时1秒）
        CompletableFuture<Double> priceFuture = CompletableFuture.supplyAsync(() -> {
            try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); }
            return 299.0; // 商品原价
        });

        // 2. 异步查询优惠券（模拟耗时0.8秒）
        CompletableFuture<Double> couponFuture = CompletableFuture.supplyAsync(() -> {
            try { Thread.sleep(800); } catch (InterruptedException e) { e.printStackTrace(); }
            return 50.0; // 优惠券金额
        });

        // 3. thenCombine合并结果：计算最终支付金额
        CompletableFuture<Double> payFuture = priceFuture.thenCombine(
                couponFuture,
                (price, coupon) -> price - coupon // 原价 - 优惠券
        );

        // 4. 输出结果（总耗时≈1秒，取最长任务时间）
        payFuture.thenAccept(finalPrice -> System.out.println("最终支付：" + finalPrice));

    }

    // CompletableFuture就像异步编程的"乐高积木"，thenApply是"串联积木"（任务依赖），thenCombine是"并联积木"（任务并行），两者结合能搭建出复杂的异步任务流。记住：
    // thenApply：链式转换，"一步接一步"
    // thenCombine：结果合并，"多步并一步"


    /**
     * 有返回值的
     */
    public void allDone2() {
        // 创建异步任务
        List<CompletableFuture<String>> futures = new ArrayList<>();
        // bo TODO 2025/8/21 17:59 注意参数传递问题,,,

        // 使用深拷贝的方式
        // QueryBaseAO baseAO = SerializationUtils.clone(data);

        for (int i = 0; i < 4; i++) {
            int finalI = i;
            futures.add(CompletableFuture.supplyAsync(() -> {
                try {
                    int millis = RandomUtil.randomInt(1000, 3000);
                    Thread.sleep(millis);
                    if (finalI == 2) {
                        // throw new RuntimeException("任务2异常");
                    }
                    return "任务" + (finalI + 1) + "完成";
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    return "任务中断";
                }
            }, threadPoolTaskExecutor));
        }

        // 组合所有任务
        CompletableFuture<Void> allFutures = CompletableFuture.allOf(
                futures.toArray(new CompletableFuture[0])
        );

        // 等待所有任务完成并处理结果
        CompletableFuture<List<String>> allResults = allFutures.thenApply(v -> {
            return futures.stream()
                    .map(CompletableFuture::join)
                    .collect(Collectors.toList());
        });

        // 获取最终结果
        try {
            List<String> strings = allResults.get();
            log.info("join:{}", strings);

        } catch (InterruptedException | ExecutionException e) {
            throw new RuntimeException(e);
        }

    }

    // 没有返回值
    public void allDone() {
        // 定义任务数量
        int taskCount = 10;
        log.info("任务开始----打印日志------{}", DateUtil.now());

        // 创建CountDownLatch，初始计数为任务数量
        CountDownLatch latch = new CountDownLatch(taskCount);


        // 提交任务
        for (int i = 0; i < taskCount; i++) {
            final int index = i;
            threadPoolTaskExecutor.submit(() -> {
                try {
                    Thread.sleep(1000);
                    System.out.println("任务 " + index + " 完成");
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                } finally {
                    latch.countDown();  // 任务完成，计数减1
                }
            });
        }

        // 等待所有任务完成
        try {
            latch.await();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }

        log.info("任务都完成-----打印日志------{}", DateUtil.now());
    }

    /**
     * 异步任务全部完成
     */
    public void allDone1() {
        log.info("任务开始-----打印日志------{}", DateUtil.now());

        // 计算耗时

        // 为每个子列表创建一个异步任务-不需要返回值的
        CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
            //     模拟业务
            ThreadUtil.sleep(1000);
        }, threadPoolTaskExecutor);


        // 为每个子列表创建一个异步任务
        CompletableFuture<Void> future2 = CompletableFuture.runAsync(() -> {
            ThreadUtil.sleep(2000);

        }, threadPoolTaskExecutor);


        CompletableFuture<Void> future3 = CompletableFuture.runAsync(() -> {
            ThreadUtil.sleep(3000);

        }, threadPoolTaskExecutor);

        // 等待所有任务完成
        CompletableFuture.allOf(future, future2, future3).join();

        log.info("任务都完成-----打印日志------{}", DateUtil.now());
    }


}
